package org.ligson.hbase127;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;

@Slf4j
public class HBase127 {

    /***
     * 10.16.25.47 crux-pp47
     * 10.16.25.26 crux-pp26
     * 10.16.25.42 crux-pp42
     * hbase.zookeeper.quorum: 127.0.0.1
     *     hbase.zookeeper.port: 2181
     *     hbase.zookeeper.znode: /hbase
     *     hbase.client.keyvalue.maxsize: 1572864000
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "10.16.25.26,10.16.25.42,10.16.25.47");
        config.set("hbase.zookeeper.port", "2181");
        //config.set("hbase.zookeeper.znode", "/hbase");
        config.set("hbase.zookeeper.znode", "/hbase2");
        Connection connection = ConnectionFactory.createConnection(config);
        HTableDescriptor[] tables = connection.getAdmin().listTables();
        for (HTableDescriptor table : tables) {
            log.debug("------：{}", table);
        }
        //connection.getAdmin().l
//        TableName tableName = TableName.valueOf("t_operation_log");
//        HTableDescriptor  ht = new HTableDescriptor(tableName);
//        HColumnDescriptor idColumn = new HColumnDescriptor("id");
//        HColumnDescriptor nameColumn = new HColumnDescriptor("name");
//        ht.addFamily(idColumn);
//        ht.addFamily(nameColumn);
//        // hcd.
//        //ht.addFamily()
//        connection.getAdmin().createTable(ht);
        createTable(connection, "t_operation_log", new String[]{"base", "extras"});
        initData(connection);


        //deleteTable(connection, "t_operation_log");
        //scan(connection, "t_operation_log");
        //Result re = scanByRowKey(connection, "t_operation_log", "ff28d235-39da-43ac-bfaa-d26e6c2e302e");
        // System.out.println("-=--=" + re);
        //List<Cell> cells = scanByFamilyAndQualifier(connection, "t_operation_log", "id", "name");
        // System.out.println(cells);

        connection.close();
    }

    public static void initData(Connection connection) throws Exception {
        List<PutRowData> putRowDataList = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            String rowKey = UUID.randomUUID().toString();

            PutRowData putRowData = new PutRowData();
            putRowData.setRowKey(rowKey);


            List<PutData> putDatas = new ArrayList<>();
            PutData idPutData = new PutData();
            idPutData.setFamily("base");
            idPutData.setQualifier("id");
            idPutData.setValue("id-" + i);
            putDatas.add(idPutData);


            PutData codePutData = new PutData();
            codePutData.setFamily("base");
            codePutData.setQualifier("code");
            codePutData.setValue("code-" + i);
            putDatas.add(codePutData);

            PutData namePutData = new PutData();
            namePutData.setFamily("base");
            namePutData.setQualifier("name");
            namePutData.setValue("name-" + i);
            putDatas.add(namePutData);

            PutData createTimePutData = new PutData();
            createTimePutData.setFamily("extras");
            createTimePutData.setQualifier("create_time");
            createTimePutData.setValue(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
            putDatas.add(createTimePutData);

            putRowData.setPutDataList(putDatas);

            putRowDataList.add(putRowData);

        }
        insertRowsValue(connection, "t_operation_log", putRowDataList);
    }

    public static void createTable(Connection connection, String tableName, String[] columnNames) throws Exception {
        TableName tn = TableName.valueOf(tableName);
        HTableDescriptor ht = new HTableDescriptor(tn);
        for (String columnName : columnNames) {
            HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(columnName);
            ht.addFamily(hColumnDescriptor);
        }
        // hcd.
        //ht.addFamily()
        connection.getAdmin().createTable(ht);
    }

    public static void deleteTable(Connection connection, String tableName) throws Exception {
        TableName tn = TableName.valueOf(tableName);
        connection.getAdmin().disableTable(tn);
        connection.getAdmin().deleteTable(tn);
    }

    public static void insertValue(Connection connection, String tableName, String rowKey, List<PutData> putDatas) throws Exception {
        TableName tn = TableName.valueOf(tableName);
        Table table = connection.getTable(tn);
        Put put = new Put(Bytes.toBytes(rowKey));
        for (PutData putData : putDatas) {
            put.addColumn(Bytes.toBytes(putData.getFamily()), Bytes.toBytes(putData.getQualifier()), Bytes.toBytes(putData.getValue()));
        }
        table.put(put);
    }

    public static void insertRowsValue(Connection connection, String tableName, List<PutRowData> putRowDataList) throws Exception {
        TableName tn = TableName.valueOf(tableName);
        Table table = connection.getTable(tn);

        List<Put> puts = new ArrayList<>();

        for (PutRowData putRowData : putRowDataList) {
            Put put = new Put(Bytes.toBytes(putRowData.getRowKey()));
            for (PutData putData : putRowData.getPutDataList()) {
                put.addColumn(Bytes.toBytes(putData.getFamily()), Bytes.toBytes(putData.getQualifier()), Bytes.toBytes(putData.getValue()));
            }
            puts.add(put);
        }
        table.put(puts);
    }

    public static void insertValue(Connection connection, String tableName, PutDatas putDatas) throws Exception {
        TableName tn = TableName.valueOf(tableName);
        Table table = connection.getTable(tn);
        Put put = new Put(Bytes.toBytes(putDatas.getRowKey()));
        for (String value : putDatas.getValues()) {
            put.addColumn(Bytes.toBytes(putDatas.getFamily()), Bytes.toBytes(putDatas.getQualifier()), Bytes.toBytes(value));
        }
        table.put(put);
    }

    public static void insertValue(Connection connection, String tableName, List<PutDatas> putDatasList) throws Exception {
        TableName tn = TableName.valueOf(tableName);
        Table table = connection.getTable(tn);
        List<Put> puts = new ArrayList<>();
        for (PutDatas putDatas : putDatasList) {
            Put put = new Put(Bytes.toBytes(putDatas.getRowKey()));
            for (String value : putDatas.getValues()) {
                put.addColumn(Bytes.toBytes(putDatas.getFamily()), Bytes.toBytes(putDatas.getQualifier()), Bytes.toBytes(value));
            }
            puts.add(put);
        }
        table.put(puts);
    }

    public static void updateValue(Connection connection, String tableName, String rowKey, List<PutData> putDatas) throws Exception {
        deleteByKey(connection, tableName, new String[]{rowKey});
        insertValue(connection, tableName, rowKey, putDatas);
    }

    public static void deleteByKey(Connection connection, String tableName, String[] keys) throws Exception {
        TableName tn = TableName.valueOf(tableName);
        Table table = connection.getTable(tn);
        List<Delete> deletes = new ArrayList<>();
        for (String key : keys) {
            Delete delete = new Delete(Bytes.toBytes(key));
            deletes.add(delete);
        }
        table.delete(deletes);
    }

    public static void scan(Connection connection, String tableName) throws Exception {
        //取得一个数据表对象
        Table table = connection.getTable(TableName.valueOf(tableName));

        //取得表中所有数据
        ResultScanner scanner = table.getScanner(new Scan());
        //循环输出表中的数据
        for (Result r : scanner) {
            byte[] row = r.getRow();
            System.out.println("row key is :" + Bytes.toString(row));
            List<Cell> listCells = r.listCells();
            for (Cell cell : listCells) {

                byte[] familyArray = CellUtil.cloneFamily(cell);
                byte[] qualifierArray = CellUtil.cloneQualifier(cell);
                // byte[] valueArray = CellUtil.cloneValue(cell);
                System.out.println("family :" + Bytes.toString(familyArray) + " qualifier:"
                        + Bytes.toString(qualifierArray));
//                System.out.println("family :" + Bytes.toString(familyArray) + " qualifier:"
//                        + Bytes.toString(qualifierArray) + " value:" + Bytes.toString(valueArray));
            }
        }
    }


    public static Result scanByRowKey(Connection connection, String tableName, String rowKey) throws Exception {
        //取得一个数据表对象
        Table table = connection.getTable(TableName.valueOf(tableName));

        //取得表中所有数据
        ResultScanner scanner = table.getScanner(new Scan());

        //循环输出表中的数据
        for (Result r : scanner) {
            byte[] row = r.getRow();
            if (ArrayUtils.isEquals(row, Bytes.toBytes(rowKey))) {
                return r;
            }
        }
        return null;
    }


    public static List<Cell> scanByFamilyAndQualifier(Connection connection, String tableName, String familyName, String qualifierName) throws Exception {
        //取得一个数据表对象
        Table table = connection.getTable(TableName.valueOf(tableName));

        //取得表中所有数据
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifierName));
        ResultScanner scanner = table.getScanner(new Scan());

        List<Cell> cells = new ArrayList<>();
        //循环输出表中的数据
        for (Result r : scanner) {
            byte[] row = r.getRow();
            System.out.println("row key is :" + Bytes.toString(row));
            List<Cell> listCells = r.listCells();
            for (Cell cell : listCells) {

                byte[] familyArray = CellUtil.cloneFamily(cell);
                byte[] qualifierArray = CellUtil.cloneQualifier(cell);
                //byte[] valueArray = CellUtil.cloneValue(cell);
                if (ArrayUtils.isEquals(familyArray, Bytes.toBytes(familyName))
                        && ArrayUtils.isEquals(qualifierArray, Bytes.toBytes(qualifierName))) {
                    cells.add(cell);
                }
            }
        }
        return cells;
    }
}
